package rpc

import (
	"context"
	"encoding/json"
	"fmt"
	"gbox-one-machine/log"
	"github.com/pkg/errors"
	"net/url"
	"reflect"
	"sync"
	"time"
	"github.com/sony/sonyflake"
	"github.com/gorilla/websocket"
)

// consts
const (
	JSONRPCVersion   = "2.0"
	DataTypeRequest  = 1
	DataTypeResponse = 2
	NullID           = 0

	MethodPing = "ping"

	StatusStarted = 0
	StatusClosed  = 1

	CallTypeRequest = 1
	CallTypeNotice  = 2
)

// error code
const (
	ErrCodeJSONRPCParseError     = -32700 // Parse error: Invalid JSON was received by the server
	ErrCodeJSONRPCInvalidRequest = -32600 // Invalid Request: The JSON sent is not a valid Request object
	ErrCodeJSONRPCMethodNotFound = -32601 // Method not found: The method does not exist / is not available
	ErrCodeJSONRPCInvalidParams  = -32602 // Invalid Params: Invalid method parameter(s)
	ErrCodeJSONRPCInternal       = -32603 // Internal error: Internal JSON-RPC error
	ErrCodeJSONRPCTimeout        = -32001 // Request timeout: The JSON-RPC request timeout
)

var chDataBuf = 512

// Errors
var (
	ErrConnClosed = errors.New("JSONRPC is closed")
)

// JSONRPCRequest defination
type JSONRPCRequest struct {
	JSONRPC string      `json:"jsonrpc"`
	ID      uint64      `json:"id"`
	Method  string      `json:"method"`
	Params  interface{} `json:"params,omitempty"`
}

func (r *JSONRPCRequest) String() string {
	reqJSON, _ := json.Marshal(r)

	return string(reqJSON)
}

// Validate request
func (r *JSONRPCRequest) Validate() error {
	if r.JSONRPC != JSONRPCVersion {
		return fmt.Errorf("Invalid Request: Version %s invalid, not %s", r.JSONRPC, JSONRPCVersion)
	}

	if r.Method == "" {
		return fmt.Errorf("Invalid Request: field method is empty")
	}

	return nil
}

// JSONRPCNotice defination
type JSONRPCNotice struct {
	JSONRPC string      `json:"jsonrpc"`
	Method  string      `json:"method"`
	Params  interface{} `json:"params,omitempty"`
}

func (n *JSONRPCNotice) String() string {
	noticeJSON, _ := json.Marshal(n)

	return string(noticeJSON)
}

// Validate request
func (n *JSONRPCNotice) Validate() error {
	if n.JSONRPC != JSONRPCVersion {
		return fmt.Errorf("Invalid Request: Version %s invalid, not %s", n.JSONRPC, JSONRPCVersion)
	}

	return nil
}

// JSONRPCResponse defination
type JSONRPCResponse struct {
	JSONRPC string        `json:"jsonrpc"`
	ID      uint64        `json:"id"`
	Result  interface{}   `json:"result,omitempty"`
	Error   *JSONRPCError `json:"error,omitempty"`
}

func (r *JSONRPCResponse) String() string {
	respJSON, _ := json.Marshal(r)

	return string(respJSON)
}

// Validate request
func (r *JSONRPCResponse) Validate() error {
	if r.JSONRPC != JSONRPCVersion {
		return fmt.Errorf("Invalid Request: Version %s invalid, not %s", r.JSONRPC, JSONRPCVersion)
	}

	if r.Result != nil && r.Error != nil {
		return fmt.Errorf("Invalid Request: field result and error are both existed")
	}

	if r.Result == nil && r.Error == nil {
		return fmt.Errorf("Invalid Request: field result and error are both not exist")
	}

	return nil
}

// JSONRPCError defination
type JSONRPCError struct {
	Code    int64       `json:"code"`
	Message string      `json:"message"`
	Data    interface{} `json:"data,omitempty"`
}

func (e *JSONRPCError) Error() string {
	return fmt.Sprintf("ErrCode: %d, ErrMsg: %s, data: %v", e.Code, e.Message, e.Data)
}

// ServiceFunc for Register
type ServiceFunc func(ctx context.Context, request interface{}) (interface{}, error)

// JSONRPCOpts for new json-rpc
type JSONRPCOpts struct {
	Ctx               context.Context
	Ws                *websocket.Conn
	URL               *url.URL
	Timeout           time.Duration
	HeartBeatInterval time.Duration
	CloseHandler      func() error
}

// JSONRPC defination, websocket conn wrap with json-rpc
type JSONRPC struct {
	ctx               context.Context
	status            int
	cancelFunc        context.CancelFunc
	closeHandler      func() error
	ws                *websocket.Conn
	id                string // ID is connection unique describe with local addr and remote addr
	chSend            chan *JSONRPCRequest
	chRecv            chan *JSONRPCRequest
	chSendNotice      chan *JSONRPCNotice
	chRecvNotice      chan *JSONRPCNotice
	chSendResponse    chan *JSONRPCResponse
	sendRespMap       sync.Map
	timeout           time.Duration
	heartBeatInterval time.Duration
	idGen             *sonyflake.Sonyflake
	serviceFuncMap    sync.Map
	sync.RWMutex
}

// NewJSONRPC create json-rpc conn
func NewJSONRPC(opts *JSONRPCOpts) (*JSONRPC, error) {

	if opts.Ctx == nil {
		opts.Ctx = context.Background()
	}

	if opts.Timeout == 0 {
		opts.Timeout = time.Second * 5
	}

	if opts.HeartBeatInterval == 0 {
		opts.HeartBeatInterval = time.Second * 20
	}

	rpc := &JSONRPC{
		status:            StatusStarted,
		chSend:            make(chan *JSONRPCRequest, chDataBuf),
		chRecv:            make(chan *JSONRPCRequest, chDataBuf),
		chSendNotice:      make(chan *JSONRPCNotice, chDataBuf),
		chRecvNotice:      make(chan *JSONRPCNotice, chDataBuf),
		chSendResponse:    make(chan *JSONRPCResponse, chDataBuf),
		timeout:           opts.Timeout,
		heartBeatInterval: opts.HeartBeatInterval,
		idGen:             sonyflake.NewSonyflake(sonyflake.Settings{}),
		closeHandler:      opts.CloseHandler,
	}

	rpc.ctx, rpc.cancelFunc = context.WithCancel(opts.Ctx)

	if opts.Ws != nil { // websocket already existed
		rpc.ws = opts.Ws
	} else if opts.URL != nil { // create websocket connection
		ws, _, err := websocket.DefaultDialer.Dial(opts.URL.String(), nil)
		if err != nil {
			return nil, err
		}

		rpc.ws = ws
	}

	rpc.id = fmt.Sprintf("%s -- %s", rpc.ws.LocalAddr(), rpc.ws.RemoteAddr())

	go rpc.recv()
	go rpc.handleRecv()
	go rpc.handleSend()
	go rpc.ping()

	return rpc, nil
}

// Name for jsonrpc
func (rpc *JSONRPC) Name() string {
	return rpc.id
}

// Call remote method, wait response
func (rpc *JSONRPC) Call(callType int, method string, args interface{}, reply interface{}) error {
	switch callType {
	case CallTypeRequest:
		if reply == nil {
			return fmt.Errorf("Reply is empty")
		}

		replyType := reflect.TypeOf(reply)
		if replyType.Kind() != reflect.Ptr {
			return fmt.Errorf("Reply type is not a pointer: %q", replyType)
		}

		resp, err := rpc.Send(method, args)
		if err != nil {
			return err
		}

		if err := json.Unmarshal(resp, reply); err != nil {
			return err
		}

		return nil
	case CallTypeNotice:
		rpc.SendNotice(method, args)
		return nil
	}
	return fmt.Errorf("Call type %d not defined", callType)
}

// Register service func to rpc server
func (rpc *JSONRPC) Register(method string, serviceFunc ServiceFunc) error {
	if _, ok := rpc.serviceFuncMap.Load(method); ok {
		return fmt.Errorf("method %s already registed", method)
	}

	rpc.serviceFuncMap.Store(method, serviceFunc)
	return nil
}

// Send start json-rpc request, wait response
func (rpc *JSONRPC) Send(method string, data interface{}) ([]byte, error) {

	req := &JSONRPCRequest{
		JSONRPC: JSONRPCVersion,
		ID:      rpc.GenID(),
		Method:  method,
		Params:  data,
	}

	resp, err := rpc.send(req)
	if err != nil {
		return nil, err
	}

	resultJSON, err := json.Marshal(resp.Result)
	if err != nil {
		return nil, err
	}

	return resultJSON, nil
}

func (rpc *JSONRPC) send(req *JSONRPCRequest) (*JSONRPCResponse, error) {

	chResp := rpc.newRespChan(req.ID)
	defer rpc.delRespChan(req.ID)

	rpc.chSend <- req

	select {
	case resp := <-chResp:
		if resp.Error != nil {
			return nil, resp.Error
		}

		return resp, nil

	case <-rpc.ctx.Done():
		return nil, ErrConnClosed

	case <-time.After(rpc.timeout):
		return nil, &JSONRPCError{
			Code:    ErrCodeJSONRPCTimeout,
			Message: fmt.Sprintf("Request timeout: The JSON-RPC request timeout after %s", rpc.timeout),
		}
	}
}

func (rpc *JSONRPC) sendResponse(resp *JSONRPCResponse) {
	rpc.chSendResponse <- resp
}

func (rpc *JSONRPC) handleSend() {
	for {
		select { // quit first
		case <-rpc.ctx.Done():
			return
		default:
		}

		var data interface{}

		select {
		case <-rpc.ctx.Done():
			return
		case req := <-rpc.chSend:
			data = req
			log.Logger.Debugf("[JSON-RPC] conn: %s send request:\n%s", rpc.id, data)
		case notice := <-rpc.chSendNotice:
			data = notice
			log.Logger.Debugf("[JSON-RPC] conn: %s send notice:\n%s", rpc.id, data)
		case resp := <-rpc.chSendResponse:
			data = resp
			log.Logger.Debugf("[JSON-RPC] conn: %s send response:\n%s", rpc.id, data)
		}

		if err := rpc.ws.WriteJSON(data); err != nil {
			log.Logger.Errorf("Message %s send err: %s\n", data, err)
			rpc.Close()
			return
		}
	}
}

// SendNotice send json-rpc notice
func (rpc *JSONRPC) SendNotice(method string, data interface{}) {

	notice := &JSONRPCNotice{
		JSONRPC: JSONRPCVersion,
		Method:  method,
		Params:  data,
	}

	rpc.chSendNotice <- notice
}

func (rpc *JSONRPC) newRespChan(id uint64) <-chan *JSONRPCResponse {
	chResp := make(chan *JSONRPCResponse, 1)
	rpc.sendRespMap.Store(id, chResp)

	return chResp
}

func (rpc *JSONRPC) getRespChan(id uint64) chan<- *JSONRPCResponse {
	if chResp, ok := rpc.sendRespMap.Load(id); ok {
		rpc.sendRespMap.Delete(id)

		ch, ok := chResp.(chan *JSONRPCResponse)
		if ok {
			return ch
		}
		return nil
	}

	return nil
}

func (rpc *JSONRPC) delRespChan(id uint64) {
	if _, ok := rpc.sendRespMap.Load(id); ok {
		rpc.sendRespMap.Delete(id)
	}
}

func (rpc *JSONRPC) recv() {

	for {
		select { // quit first
		case <-rpc.ctx.Done():
			return
		default:
		}

		select {
		case <-rpc.ctx.Done():
			return
		case req := <-rpc.chRecv:
			go rpc.serveRequest(req)
		case notice := <-rpc.chRecvNotice:
			go rpc.serveNotice(notice)
		}
	}
}

func (rpc *JSONRPC) serveRequest(req *JSONRPCRequest) {
	resp := &JSONRPCResponse{
		JSONRPC: JSONRPCVersion,
		ID:      req.ID,
	}
	iServiceFunc, ok := rpc.serviceFuncMap.Load(req.Method)
	if !ok {
		resp.Error = &JSONRPCError{
			Code:    ErrCodeJSONRPCMethodNotFound,
			Message: fmt.Sprintf("Request Method %s not found", req.Method),
		}
		rpc.sendResponse(resp)
		return
	}
	serviceFunc := iServiceFunc.(ServiceFunc)
	paramsJSON, err := json.Marshal(req.Params)
	if err != nil {
		resp.Error = &JSONRPCError{
			Code:    ErrCodeJSONRPCInvalidParams,
			Message: fmt.Sprintf("Invalid Params %s", req.Params),
		}
		rpc.sendResponse(resp)
		return
	}
	result, err := serviceFunc(rpc.ctx, paramsJSON)
	if err != nil {

		if errJSONRPC, ok := err.(*JSONRPCError); ok {
			resp.Error = errJSONRPC
		} else {
			resp.Error = &JSONRPCError{
				Code:    ErrCodeJSONRPCInternal,
				Message: err.Error(),
			}
		}

		rpc.sendResponse(resp)
		return
	}

	resp.Result = result
	rpc.sendResponse(resp)

}

func (rpc *JSONRPC) serveNotice(notice *JSONRPCNotice) {
	iServiceFunc, ok := rpc.serviceFuncMap.Load(notice.Method)
	if !ok {
		log.Logger.Errorf("Notice Method %s not found", notice.Method)
		return
	}
	serviceFunc := iServiceFunc.(ServiceFunc)
	paramsJSON, err := json.Marshal(notice.Params)
	if err != nil {
		log.Logger.Errorf("Invalid Params %s", notice.Params)
		return
	}
	if _, err := serviceFunc(rpc.ctx, paramsJSON); err != nil {
		log.Logger.Errorf("Notice Method %s exec err: %s", notice.Method, err)
		return
	}
}

func (rpc *JSONRPC) handleRecv() {

	for {
		select { // quit first
		case <-rpc.ctx.Done():
			return
		default:
		}

		_, dataBuf, err := rpc.ws.ReadMessage()
		if err != nil {
			log.Logger.Errorf("Message read err: %s\n", err)
			rpc.Close()
			return
		}

		iData, err := parseJSONRPCData(dataBuf)
		if err != nil { // recv data invalid
			log.Logger.Errorf("JSON-RPC data %s parse err: %s\n", dataBuf, err)

			rpc.sendResponse(&JSONRPCResponse{
				JSONRPC: JSONRPCVersion,
				ID:      NullID,
				Error: &JSONRPCError{
					Code:    ErrCodeJSONRPCParseError,
					Message: err.Error(),
				},
			})

			continue
		}

		switch data := iData.(type) {
		case *JSONRPCRequest: // the data is request for Recv()

			log.Logger.Debugf("[JSON-RPC] conn: %s recv request:\n%s", rpc.id, data)
			rpc.handleRecvRequest(data)

		case *JSONRPCNotice: // the data is request for RecvNotice()

			log.Logger.Debugf("[JSON-RPC] conn: %s recv notice:\n%s", rpc.id, data)
			rpc.handleRecvNotice(data)

		case *JSONRPCResponse: // the data is response for send()

			log.Logger.Debugf("[JSON-RPC] conn: %s recv response:\n%s", rpc.id, data)
			rpc.handleRecvResponse(data)

		}
	}
}

func (rpc *JSONRPC) handleRecvRequest(req *JSONRPCRequest) {

	if err := req.Validate(); err != nil { // response error
		rpc.sendResponse(&JSONRPCResponse{
			JSONRPC: JSONRPCVersion,
			ID:      req.ID,
			Error: &JSONRPCError{
				Code:    ErrCodeJSONRPCInvalidRequest,
				Message: err.Error(),
			},
		})
		log.Logger.Errorf("Invalid JSON-RPC reqeust data: %s, err: %s", req, err)
		return
	}

	if req.Method == MethodPing { // for ping request
		rpc.pong(req)
	} else { // normal request
		rpc.chRecv <- req
	}
}

func (rpc *JSONRPC) handleRecvNotice(notice *JSONRPCNotice) {
	if err := notice.Validate(); err != nil {
		log.Logger.Errorf("Invalid JSON-RPC notice data: %s, err: %s", notice, err)
	}

	rpc.chRecvNotice <- notice
}

func (rpc *JSONRPC) handleRecvResponse(resp *JSONRPCResponse) {

	if chResp := rpc.getRespChan(resp.ID); chResp != nil {
		if err := resp.Validate(); err != nil { // invalid response, change response to error response
			log.Logger.Errorf("Invalid JSON-RPC response data: %s, err: %s", resp, err)
			chResp <- &JSONRPCResponse{
				JSONRPC: JSONRPCVersion,
				ID:      resp.ID,
				Error: &JSONRPCError{
					Code:    ErrCodeJSONRPCInvalidRequest,
					Message: err.Error(),
				},
			}
			return
		}

		chResp <- resp
	}
}

func (rpc *JSONRPC) ping() {

	timer := time.NewTimer(rpc.heartBeatInterval)

	for {
		select { // quit first
		case <-rpc.ctx.Done():
			return
		default:
		}

		timer.Reset(rpc.heartBeatInterval)

		select {
		case <-rpc.ctx.Done():
			return
		case <-timer.C:
			_, err := rpc.Send(MethodPing, nil)
			if err != nil {
				rpc.Close()
				return
			}
		}
	}
}

func (rpc *JSONRPC) pong(req *JSONRPCRequest) {
	resp := &JSONRPCResponse{
		JSONRPC: JSONRPCVersion,
		ID:      req.ID,
		Result:  "pong",
	}

	rpc.chSendResponse <- resp
}

// GenID generate sonwflake id
func (rpc *JSONRPC) GenID() uint64 {

	id, _ := rpc.idGen.NextID()

	return id
}

// Close json-rpc
func (rpc *JSONRPC) Close() error {

	if !rpc.setStatus(StatusClosed) { // already closed
		return nil
	}

	rpc.cancelFunc()

	if rpc.closeHandler != nil {
		go func() {
			if err := rpc.closeHandler(); err != nil {
				log.Logger.Errorf("[JSON-RPC] conn: %s run close handler err: %s", rpc.id, err)
			}
		}()
	}

	return rpc.ws.Close()
}

func (rpc *JSONRPC) setStatus(status int) bool {
	rpc.Lock()
	defer rpc.Unlock()

	if rpc.status != status {
		rpc.status = status
		return true
	}

	return false
}

func parseJSONRPCData(dataJSON []byte) (interface{}, error) {
	data := make(map[string]interface{})
	if err := json.Unmarshal(dataJSON, &data); err != nil {
		return nil, fmt.Errorf("Parse error: Invalid JSON was received by the server %s", dataJSON)
	}

	_, ok := data["id"]

	// notice data
	if !ok {
		notice := &JSONRPCNotice{}
		if err := json.Unmarshal(dataJSON, &notice); err != nil {
			return nil, fmt.Errorf("Invalid Request: %s", err)
		}

		return notice, nil
	}

	// request data
	if _, ok := data["method"]; ok {

		req := &JSONRPCRequest{}
		if err := json.Unmarshal(dataJSON, &req); err != nil {
			return nil, fmt.Errorf("Invalid Request: %s", err)
		}

		return req, nil
	}

	// response data
	resp := &JSONRPCResponse{}
	if err := json.Unmarshal(dataJSON, &resp); err != nil {
		return nil, fmt.Errorf("Invalid Request: %s", err)
	}

	return resp, nil
}
